package com.sailing.lianxi.demoservice.onecusmer;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import com.sailing.lianxi.common.Constants;

/**
 * 精确一次

下例展示了kafka的精确一次语义，consumer通过subscribe方法注册到kafka，精确一次的语义要求必须手动管理offset，按照下述步骤进行设置：
1.设置enable.auto.commit = false;
2.处理完消息之后不要手动提交offset，
3.通过subscribe方法将consumer注册到某个特定topic，
4.实现ConsumerRebalanceListener接口和consumer.seek(topicPartition,offset)方法（读取特定topic和partition的offset）
5.将offset和消息一块存储，确保原子性，推荐使用事务机制。
这里展示的示例是将offset存储到文件中，如果要存储到数据库中的话，需要修改offsetmanger类，将offset写入数据库
 
 * @author Administrator
 *
 */
public class ExactlyOnceDynamicConsumer {

	  private static OffsetManager offsetManager = new OffsetManager("storage2");
	  
	  
	  /**
	   * 消费数据
	   */
	  private static void consumerMessage() {
		  KafkaConsumer<String, String> consumer = createConsumer();
		  consumer.subscribe(Arrays.asList("topic1,topic2"),new MyConsumerRebalancerListener(consumer));
		  readMessage(consumer);
	  }
	  
	/**
	 * 创建一个消费者
	 * @return
	 */
	private static KafkaConsumer<String, String> createConsumer(){
		Properties props = new Properties();
		//kafka服务器地址  
		props.put("bootstrap.servers", Constants.KAFKA_SERVERS);  
		//kafka组
        props.put("group.id", "wg10");  
        /*
        earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 
        latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 
        none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常 
                 以上配置只对一个组里没有消费的消息起作用
        */
        props.put("auto.offset.reset", "earliest");  
        props.put("enable.auto.commit", "false");  
        //Consumer向集群发送自己的心跳，超时则认为Consumer已经死了，kafka会把它的分区分配给其他进程,默认6秒
        props.put("session.timeout.ms", "6000");//毫秒 
        //key序列化器
        String deserialClass = StringDeserializer.class.getName();
        props.put("key.deserializer", deserialClass);  
        //值序列化器
        props.put("value.deserializer", deserialClass);
        
        //设置最大拉取SIZE
        props.put("fetch.message.max.bytes",1024 *1024*100);//设置最大拉取SIZE
        props.put("max.poll.records", 1);
	     //创建消费者
	     return new KafkaConsumer<String, String>(props);
	}
	
	private static void readMessage(KafkaConsumer<String, String> consumer) {
		while(true) {
			ConsumerRecords<String, String> poll = consumer.poll(100);
			for (ConsumerRecord<String, String> record : poll) {
				 System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
				 offsetManager.saveOffsetInExternalStore(record.topic(), record.partition(), record.offset());
			}
		}
	}
	
	public static void main(String[] args) {
		consumerMessage();
	}
}
